/*
 * Copyright 2016-2021 Pnoker. All Rights Reserved.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *     http://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.xj.center.data.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.ie.api.center.manager.feign.DeviceClient;
import com.ie.api.center.manager.feign.PointClient;
import com.xj.center.data.service.ControlInfoService;
import com.xj.center.data.service.DataCustomService;
import com.xj.center.data.service.PointValueService;
import com.xj.common.bean.Pages;
import com.xj.common.bean.R;
import com.xj.common.bean.constant.Common;
import com.xj.common.bean.point.PointValue;
import com.xj.common.dto.ControlInfoDto;
import com.xj.common.dto.PointValueDto;
import com.xj.common.model.ControlInfo;
import com.xj.common.model.Device;
import com.xj.common.model.Point;
import com.xj.common.utils.JsonPaseUtil;
import com.xj.common.utils.KafkaProducer;
import com.xj.common.utils.RedisUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.index.Index;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author pnoker
 */
@Slf4j
@Service
public class ControlInfoServiceImpl implements ControlInfoService {

    @Resource
    private KafkaProducer kafkaProducer;
    @Resource
    private RedisUtil redisUtil;
    @Resource
    private ThreadPoolExecutor threadPoolExecutor;
    @Resource
    private MongoTemplate mongoTemplate;

    @Resource
    private DeviceClient deviceClient;

    @Override
    public String sendCtl(ControlInfo controlInfo) {

        if(null!= controlInfo){
            log.debug("Send controlInfo value{}", JSON.toJSONString(controlInfo));
            Long id = System.currentTimeMillis();//生成一个id
            controlInfo.setId(id.toString());
            kafkaProducer.send(Common.KAFKA.TOPIC_EXCHANGE_CONTROL_CLIENT, JsonPaseUtil.ObjToJsonStr(controlInfo));
            return id.toString();
        }
        return null;
    }

    /**
     * 存入控制信息到redis中，key为：device_control::[deviceid]   此处一个设备只能一个控制，后续需要增加控点的配置，支撑多个控制。
     * @param controlInfo
     */
    @Override
    public void saveControlInfoToRedis(ControlInfo controlInfo) {
        if(controlInfo!=null){
            //写入redis
            threadPoolExecutor.execute(() -> {
                try {
                    //rediskey:   device_control::[handleid]
                    redisUtil.setKey(
                            Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + controlInfo.getId() ,
                            controlInfo,
                            controlInfo.getClearTime(),
                            controlInfo.getTimeUnit()
                    );
                } catch (Exception e) {
                    log.error("Save point values to redis error {}", e.getMessage());
                }
            });
        }


    }

    /**
     * 更新控制实时数据 key为：device_control::[handleid]   后续更新计划：此处一个设备只能一个控制，后续需要增加控点的配置，支撑多个控制。
     * @param controlInfo
     */
    @Override
    public void updateControlInfoToRedis(ControlInfo controlInfo) {
        if(controlInfo!=null){
            //写入redis
            threadPoolExecutor.execute(() -> {
                try {
                    String key = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + controlInfo.getId();
                    //删除原有的
                    redisUtil.deleteKey(key);
                    redisUtil.setKey(
                            key,
                            controlInfo,
                            controlInfo.getClearTime(),
                            controlInfo.getTimeUnit()
                    );
                } catch (Exception e) {
                    log.error("Save point values to redis error {}", e.getMessage());
                }
            });
        }
    }

    //存放历史数据，只有成功或者失败才会保存进入mongo
    @Override
    public void saveControlInfoToMongo(ControlInfo controlInfo) {
        if(controlInfo != null){
            //写入mongodb
            threadPoolExecutor.execute(() -> {
                try {
                    String collection = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + controlInfo.getDeviceId();
                    ensureControlInfoIndex(collection);
                    mongoTemplate.insert(controlInfo, collection);
                } catch (Exception e) {
                    log.error("Save point values to mongo error {}", e.getMessage());
                }
            });
        }
    }
    /**
     * Ensure device point & time index
     *
     * @param collection Collection Name
     */
    public void ensureControlInfoIndex(String collection) {
        // ensure time index
        Index timeIndex = new Index();
        timeIndex.background().on("reqtime", Sort.Direction.DESC).named("IX_time");
        mongoTemplate.indexOps(collection).ensureIndex(timeIndex);
    }

    /**
     * 获取实时控制下发结果信息  已弃用
     * @param deviceId Device Id
     * @return
     */
    @Override
    public ControlInfo realtimeByDeviceId(Long deviceId) {
        ControlInfo controlInfo = null ;
       /* String collection = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + deviceId;
        Criteria criteria = new Criteria();
        Query query = new Query(criteria);
        query.with(Sort.by(Sort.Direction.DESC, "reqtime"));
        controlInfo = mongoTemplate.findOne(query, ControlInfo.class, collection);*/

        return controlInfo;
    }

    //获取控制结果
    @Override
    public ControlInfo realtimeByCtlId(String handle) {
        String key = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + handle;
        ControlInfo  controlInfo = redisUtil.getKey(key , ControlInfo.class);
        if (null != controlInfo) {
            controlInfo.setTimeOut(null);
            controlInfo.setTimeUnit(null);
        }
        return controlInfo;
    }

    @Override
    public List<ControlInfo> latest(Long deviceId) {
        List<ControlInfo> lcontrolInfo = new ArrayList<>();
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("deviceId",deviceId);
        R<Device> deviceR = deviceClient.selectById(jsonObject);
        if (!deviceR.isOk()) {
            return null;
        }
        String collection = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + deviceR.getData().getId();
        Criteria criteria = new Criteria();
        criteria.and("deviceId").is(deviceR.getData().getId());
        Query query = new Query(criteria);
        query.with(Sort.by(Sort.Direction.DESC, "reqtime"));//按照请求时间进行排序
        lcontrolInfo.add(mongoTemplate.findOne(query, ControlInfo.class, collection));
        return lcontrolInfo;
    }

    @Override
    @SneakyThrows
    public Page<ControlInfo> list(ControlInfoDto controlInfoDto) {
        Criteria criteria = new Criteria();
        controlInfoDto = Optional.ofNullable(controlInfoDto).orElse(new ControlInfoDto());
        if (null != controlInfoDto.getDeviceId()) {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("deviceId",controlInfoDto.getDeviceId());
            R<Device> deviceR = deviceClient.selectById(jsonObject);
            if (deviceR.isOk()) {
                Device device = deviceR.getData();
                criteria.and("deviceId").is(controlInfoDto.getDeviceId());
            }
        }
        Pages pages = Optional.ofNullable(controlInfoDto.getPage()).orElse(new Pages());
        if (pages.getStartTime() > 0 && pages.getEndTime() > 0 && pages.getStartTime() <= pages.getEndTime()) {
            criteria.and("reqtime").gte(new Date(pages.getStartTime())).lte(new Date(pages.getEndTime()));
        }
        String collection = Common.Cache.REAL_TIME_CONTROL_KEY_PREFIX + controlInfoDto.getDeviceId();
        Future<Long> count = threadPoolExecutor.submit(() -> {
            Query query = new Query(criteria);
            return mongoTemplate.count(query, ControlInfo.class, collection);
        });

        Future<List<ControlInfo>> controlInfos = threadPoolExecutor.submit(() -> {
            Query query = new Query(criteria);
            query.limit((int) pages.getSize()).skip(pages.getSize() * (pages.getCurrent() - 1));
            query.with(Sort.by(Sort.Direction.DESC, "reqtime"));
            return mongoTemplate.find(query, ControlInfo.class, collection);
        });

        Page<ControlInfo> controlInfoPage = new Page<>();
        List<ControlInfo> values = controlInfos.get().stream().filter(Objects::nonNull).collect(Collectors.toList());
        controlInfoPage.setCurrent(pages.getCurrent()).setSize(pages.getSize()).setTotal(count.get()).setRecords(values);

        return controlInfoPage;
    }
}
